/*
	Created by Jingle Wang 

*/

#ifndef TKDATABASE_CLIENT_H
#define TKDATABASE_CLIENT_H

#include "../../utils/include/utils.h"
#include "../../utils/include/msg_queue.h"
#include "../../consensus/include/tk_message.h"
#include "../../transporter/include/rdma_transporter.h"
#include "../../serialization/include/tk_jute.h"
#include <vector>
#include <string>
#include <array>
#include <thread>
#include <chrono>
#include <ctime>
#include <unordered_map>
#include <gtest/gtest.h>
#define CONFLICT_KEY "/CONFLICT"
#define TEST_KEY "/TEST"

void waitReplies(std::vector<int> & servers, int leader, int n, MsgQueue * mq);

void wait_replies_rdma(rdma_fd *handler, int leader, int n, MsgQueue *mq);
int build_connection(rdma_fd *handler, const std::string & remote_ip, uint16_t port);

int client_send_message(rdma_fd *handler, serializeBuff *message, TYPE msg_type);

void prepare_latency_data_no_conflict(std::vector<serializeBuff *> &data, int count, int valsize);

void wait_replies_for_latency_test(rdma_fd *handler);

int build_connection(rdma_fd *handler, const std::string & remote_ip, uint16_t port){
    int ret = 0;
    int master = dialTo(remote_ip, port);
    if(master < 0){
        fprintf(stdout, "Error connecting to master !\n");
        ret = -1;
        return ret; 
    }
    uint8_t msgType;
    GetReplicaListArgs rl;
    GetReplicaListReply rlp;
    rl.Marshal(master);
    if (readUntil(master, (char *) &msgType, 1) != 0 || (TYPE)msgType != GET_REPLICA_LIST_REPLY ||
        !rlp.Unmarshal(master)) {
        fprintf(stdout, "Error when getting replica lists from master!\n");
        ret = -1;
        return ret;
    }
    if (!rlp.Ready) {
        fprintf(stdout, "Master's reply shows that replicas are not ready!\n");
        ret = -1;
        return ret;
    }
    int conn = dialTo(rlp.ReplicaAddrList[0], (uint16_t )rlp.ReplicaPortList[0]);
    if(conn < 0){
        fprintf(stdout, "Error connecting to replica %d\n", 0);
        ret = -1;
        return ret;
    }
    context_info *rdma_ib_info = (context_info *)malloc(sizeof(context_info));
    open_device_and_alloc_pd(rdma_ib_info);
    handler->fd = conn;               
    get_context_info(handler, rdma_ib_info);
    build_rdma_connection(handler);
    printf("connection complete!\n");
    return ret;
}

int client_send_message(rdma_fd *handler, serializeBuff *message, TYPE msg_type){
    int ret = 0;
//    query_qp(handler);
    int send_buf_size = sizeof(uint8_t) + sizeof(uint32_t) + 1 + message->off + 1;
    char *send_buf = (char *)malloc(send_buf_size);
    gen_send_buf(send_buf, message, msg_type);
    ret = rdma_write(handler, send_buf, send_buf_size);
    free(send_buf);
    return ret;
}


void prepare_latency_data_no_conflict(std::vector<serializeBuff *> &data, int count, int valsize){
	//std::vector<std::string> karray(count, "");
    std::string _bufBegin = randStr(valsize, valsize);
    char * buf = new char[valsize];
    memset(buf, 0, (size_t)valsize);
    strcpy(buf, _bufBegin.c_str());
    auto start = std::chrono::system_clock::now();
	for(int i = 0; i < count; i++){
		std::string key = TEST_KEY + std::to_string(i);
		tk_command tkc;
    	Propose args(i, 0, tkc, -1);
    	//args.CommandId = id;
        args.Command.opcode = PUT;
        args.Command.val = buf;
        args.Command.valSize = valsize;
        args.Command.key = key;
        serializeBuff *message = create_serialize_buff();
        serializePropose(&args, message);
        data[i] = message;
        //args.Command.valSize = randRange(FLAGS_mins, FLAGS_maxs);
	}
	std::chrono::system_clock::time_point end = std::chrono::system_clock::now();
	std::chrono::duration<double> diff = end-start;
    printf("[Serialization TIME]: %lfs [Latency]: %lf us\n", diff.count(), (double) diff.count() * 1000000 / count);
    //Propose *propse = new Propose();
    auto start1 = std::chrono::system_clock::now();
	for(int i = 0; i < count; i++){
		Propose *propose = new Propose();        
        deserializePropose(data[i], propose);
        delete propose;
        //args.Command.valSize = randRange(FLAGS_mins, FLAGS_maxs);
	}
	auto end1 = std::chrono::system_clock::now();
	std::chrono::duration<double> diff1 = end1-start1;
    printf("[Deserialization TIME]: %lfs [Latency]: %lf us\n", diff1.count(), (double) diff1.count() * 1000000 / count);

}


void wait_replies_for_latency_test(rdma_fd *handler){
//    ProposeReplyTS reply;
    uint8_t msg_type = 0;
    //serializeBuff *message;
    int len = 0;
    //message->buff = NULL;
    if(handler->have_read + sizeof(uint8_t) + sizeof(uint32_t) + 1 > handler->receive_buf_size){
        handler->have_read = 0;
    }
    while(1){
        volatile char *end = (volatile char *)(handler->receive_buf + handler->have_read + 5);
        if(*end == '1'){
            break;
        }
    }
    memcpy(&msg_type, handler->receive_buf + handler->have_read, sizeof(uint8_t));
    memcpy(&len, handler->receive_buf + handler-> have_read + sizeof(uint8_t), sizeof(uint32_t));
    //msg_type = *((uint8_t *)handler->receive_buf + handler->have_read);
    //message->len = *((uint32_t *)handler->receive_buf + handler->have_read + sizeof(uint8_t));
    //printf("1.have receive msg from id: %d, msg len: %d\n", remote_id, message->len);
    memset(handler->receive_buf + handler->have_read, 0, sizeof(uint8_t) + sizeof(uint32_t) + 1);
    handler->have_read += sizeof(uint8_t) + sizeof(uint32_t) + 1;
    if(handler->have_read + len + 1 > handler->receive_buf_size){
        handler->have_read = 0;
    }
    while(1){
        volatile char *end = (volatile char *)(handler->receive_buf + handler->have_read + len);
        if(*end == '1'){
            break;  
        }
    }
    //printf("2.have receive msg from id: %d, msg len: %d\n", remote_id, message->len);
    //char * buff = (char * )malloc(len);
    //memcpy(buff, handler->receive_buf + handler->have_read, len);
    memset(handler->receive_buf + handler->have_read, 0, len + 1);
    //message = create_serialize_buff(buff,len);
    handler->have_read += len + 1;
//        printf("before dispatch, message: %p, message->len: %d\n", message, message->len);
//        r->msg_dispatcher(remote_id, message, (TYPE)msg_type);
    /*int ret = 0;
    ProposeReply *propose_reply = new ProposeReply();
    ProposeReplyTS *propose_reply_ts = new ProposeReplyTS();
    if((TYPE)msg_type == PROPOSE_REPLY){
        ret = deserializeProposeReply(message, propose_reply);
        if(ret == -1){
            return;
        }
    }
    else if((TYPE)msg_type == PROPOSE_REPLY_TS){
        ret = deserializeProposeReplyTS(message, propose_reply_ts);
        if(ret == -1){
            return;
        }
    }
    delete propose_reply;
    delete propose_reply_ts;*/
    //delete message;
}






#endif //TKDATABASE_CLIENT_H
